package cn.tukk.sparkproject.spark.page;


import cn.tukk.sparkproject.constant.Constants;
import cn.tukk.sparkproject.dao.IPageSplitConvertRateDAO;
import cn.tukk.sparkproject.dao.ITaskDAO;
import cn.tukk.sparkproject.dao.factory.DAOFactory;
import cn.tukk.sparkproject.domain.PageSplitConvertRate;
import cn.tukk.sparkproject.domain.Task;
import cn.tukk.sparkproject.util.DateUtils;
import cn.tukk.sparkproject.util.NumberUtils;
import cn.tukk.sparkproject.util.ParamUtils;
import cn.tukk.sparkproject.util.SparkUtils;
import com.alibaba.fastjson.JSONObject;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.IntegerType;
import scala.Tuple2;

import java.util.*;

/**
 * 页面跳转率模块 spark作业
 */
public class PageOneStepConvertRateSpark {
	public static void main(String[] args) {
		//1、构建Spark上下文
		SparkConf conf = new SparkConf()
				.setAppName(Constants.SPARK_APP_NAME_PAGE);
		SparkUtils.setMaster(conf);
		JavaSparkContext sc = new JavaSparkContext(conf);
		SQLContext sqlContext = SparkUtils.getSQLContext(sc.sc());

		//2、生成模拟数据
		SparkUtils.mockData(sc,sqlContext);
		//3、查询任务，获取任务参数
		Long taskid = ParamUtils.getTaskIdFromArgs(args, Constants.SPARK_LOCAL_TASKID_PAGE);
		ITaskDAO taskDAO = DAOFactory.getTaskDAO();
		Task task = taskDAO.findById(taskid);
		if(task==null){
			System.out.println(new Date()+":cannot find this task with id["+taskid+"].");
			return;
		}
		JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());

		//4、查询指定日期范围内的用户访问行为数据
		JavaRDD<Row> actionRDD = SparkUtils.getActionRDDByDateRange(sqlContext, taskParam);

		//拿到用户行为rdd后，对用户访问行为数据做一个映射，将其映射为<sessionid,访问方式>的格式
		//页面切片 是基于session粒度，从页面到不同页面之间转化  比如A用户访问了  页面3和页面5  用户b访问了页面4和页面6
		JavaPairRDD<String,Row> sessionid2actionRDD=getSessionid2actionRDD(actionRDD);

		sessionid2actionRDD = sessionid2actionRDD.cache();//Persist(StorageLevel.MEMORY_ONLY)

		//<sessionid,访问行为> RDD，做一次groupbykey操作   ,拿到每个session对应的访问行为数据，才能生成切片
		JavaPairRDD<String, Iterable<Row>> sessionid2actionsRDD = sessionid2actionRDD.groupByKey();

		//最核心的一步，每个session的单跳页面切片的生成，以及页面流的匹配，算法  sc为了广播变量
		JavaPairRDD<String,Integer> pageSplitRDD=generateAndMatchPageSplit(sc,sessionid2actionsRDD,taskParam);

		Map<String, Object> pageSplitPvMap = pageSplitRDD.countByKey();//是一个action算子 countByKey用于统计RDD[K,V]中每个K的数量

		//使用者指定页面流是3 ，2,5,8,6
		//咱们现在拿到的是pageSplitPvMap 3->2,2->5,5->8,8->6
		long startPagePv=getStartPagePv(taskParam,sessionid2actionsRDD);

		
		//计算目标页面流的各个页面切片转化率
		Map<String,Double> convertRateMap=computePageSplitConvertRate(taskParam,pageSplitPvMap,startPagePv);
		
		//持久化页面切片转化率
		persistConvertRate(taskid,convertRateMap);

	}

	/**
	 * 持久化转化率
	 * @param taskid
	 * @param convertRateMap
	 */
	private static void persistConvertRate(Long taskid, Map<String,Double> convertRateMap) {
		StringBuffer buffer = new StringBuffer("");
		
		for (Map.Entry<String,Double> convertRateEntry : convertRateMap.entrySet()) {
			String pageSplit = convertRateEntry.getKey();
			Double convertRate = convertRateEntry.getValue();
			buffer.append(pageSplit+"="+convertRate+"|");
		}
		String convertRate = buffer.toString();
		if(convertRate.length()>1) {
			convertRate = convertRate.substring(0, convertRate.length() - 1);
		}

		PageSplitConvertRate pageSplitConvertRate = new PageSplitConvertRate();
		pageSplitConvertRate.setTaskid(taskid);
		pageSplitConvertRate.setConvertRate(convertRate);

		IPageSplitConvertRateDAO pageSplitConvertRateDAO = DAOFactory.getPageSplitConvertRateDAO();
		pageSplitConvertRateDAO.insert(pageSplitConvertRate);

	}

	/**
	 * 计算页面持久化转化率
	 * @param taskParam
	 * @param pageSplitPvMap
	 * @param startPagePv
	 * @return
	 */
	private static Map<String,Double> computePageSplitConvertRate(
			JSONObject taskParam,
			Map<String,Object> pageSplitPvMap,
			long startPagePv) {
		HashMap<String, Double> convertRateMap = new HashMap<>();
		String[] targetPages = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW).split(",");
		long lastPageSplitPv=0L;
		// 3,5,2,4,6
		// 3_5
		// 3_5 pv / 3 pv
		// 5_2 rate = 5_2 pv / 3_5 pv
		for (int i=1;i<targetPages.length;i++){
			String targetPageSplit = targetPages[i - 1] + "_" + targetPages[i];
			Long targetPageSplitPv = Long.valueOf(String.valueOf(pageSplitPvMap.get(targetPageSplit)));
			double convertRate=0.0;
			if(i==1){
				convertRate=NumberUtils.formatDouble((double)targetPageSplitPv/(double)startPagePv,2);//
			}else {
				convertRate = NumberUtils.formatDouble((double) targetPageSplitPv / (double) lastPageSplitPv, 2);
			}
			convertRateMap.put(targetPageSplit,convertRate);
			lastPageSplitPv=targetPageSplitPv;
		}

		return convertRateMap;
	}

	/**
	 * 获取页面流中出事页面的pv
	 * @param taskParam
	 * @param sessionid2actionsRDD
	 * @return
	 */
	private static long getStartPagePv(JSONObject taskParam, JavaPairRDD<String,Iterable<Row>> sessionid2actionsRDD) {

		String targetPageFlow = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW);
		final long startPageId = Long.valueOf(targetPageFlow.split(",")[0]);
		JavaRDD<Long> startPageRDD = sessionid2actionsRDD.flatMap(new FlatMapFunction<Tuple2<String, Iterable<Row>>, Long>() {
			private static final long serialVersionUID = 782633779768983003L;
			@Override
			public Iterable<Long> call(Tuple2<String, Iterable<Row>> tuple2) throws Exception {
				List<Long> list = new ArrayList<Long>();
				Iterator<Row> iterator = tuple2._2.iterator();
				while (iterator.hasNext()){
					Row row = iterator.next();
					long pageid = row.getLong(3);
					if(pageid==startPageId){
						list.add(pageid);
					}
				}
				return list;
			}
		});

		return startPageRDD.count();
	}

	/**
	 * 获取<sessionid2,用户行为>  格式数据
	 * @param sc
	 * @param sessionid2actionsRDD
	 * @param taskParam
	 * @return
	 */
	private static JavaPairRDD<String,Integer> generateAndMatchPageSplit(
			JavaSparkContext sc,
			JavaPairRDD<String,Iterable<Row>> sessionid2actionsRDD,
			JSONObject taskParam) {
		String targetPageFlow = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW);
		final Broadcast<String>  targetPageFlowBroadCast=sc.broadcast(targetPageFlow);

		return sessionid2actionsRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<String, Iterable<Row>>, String, Integer>() {

			private static final long serialVersionUID = -9169137396859756857L;

			@Override
			public Iterable<Tuple2<String, Integer>> call(Tuple2<String, Iterable<Row>> tuple2) throws Exception {
				List<Tuple2<String, Integer>> returnList = new ArrayList<>();
				//获取当前session访问行为的迭代器
				Iterator<Row> ite = tuple2._2.iterator();
				//获取使用者指定页面刘   指定页面流比如1,2,3,4,5,6,7
				//1-->2的转化率是多少？2-->3的转化率是多少
				String[] targetPages = targetPageFlowBroadCast.value().split(",");

				//拿到的session访问行为，时间是乱序的，但是希望按照时间排序  那么第一件事情就是按照时间排序
				ArrayList<Row> rows = new ArrayList<>();
				while(ite.hasNext()){
					rows.add(ite.next());
				}
				Collections.sort(rows, new Comparator<Row>() {
					@Override
					public int compare(Row o1, Row o2) {
						String actionTime1 = o1.getString(4);
						String actionTime2 = o2.getString(4);

						Date date1 = DateUtils.parseTime(actionTime1);
						Date date2 = DateUtils.parseTime(actionTime2);
						return (int)(date1.getTime()-date2.getTime());//时间升序从 比如2018-12 到2019-01
					}
				});

				//页面切片的生成，以及页面流的匹配
				Long lastPageId =null;
				for (Row row : rows) {
					long pageid = row.getLong(3);
					if(lastPageId==null){
						lastPageId=pageid;
						continue;
					}

					// 生成一个页面切片
					// 3,5,2,1,8,9
					// lastPageId=3
					// 5，切片，3_5

					String pageSplit = lastPageId + "_" + pageid;
					//对于这个切片判断，是否在用指定页面流中
					for (int i = 1; i <targetPages.length ; i++){
						String targetPageSplit = targetPages[i - 1] + "_" + targetPages[i];

						if(pageSplit.equals(targetPageSplit)){
							returnList.add(new Tuple2<>(pageSplit,1));
							break;
						}
					}
					lastPageId=pageid;
				}


				return returnList;
			}
		});
	}



	/**
	 * 获取<sessionid,用户访问行为>格式数据
	 * @param actionRDD
	 * @return
	 */
	private static JavaPairRDD<String,Row> getSessionid2actionRDD(JavaRDD<Row> actionRDD) {
		return  actionRDD.mapToPair(new PairFunction<Row, String, Row>() {

			private static final long serialVersionUID = -7399496367546671791L;

			@Override
			public Tuple2<String, Row> call(Row row) throws Exception {
				String sessionid=row.getString(2);
				return new Tuple2<>(sessionid,row);
			}
		});
	}

}
































